package com.example.zyxdemo.config;

import jakarta.annotation.Resource;
import org.springframework.boot.autoconfigure.kafka.KafkaProperties;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.kafka.annotation.EnableKafka;
import org.springframework.kafka.config.ConcurrentKafkaListenerContainerFactory;
import org.springframework.kafka.config.KafkaListenerContainerFactory;
import org.springframework.kafka.core.ConsumerFactory;
import org.springframework.kafka.core.DefaultKafkaConsumerFactory;
import org.springframework.kafka.listener.ConcurrentMessageListenerContainer;
import org.springframework.kafka.listener.ContainerProperties;



/**
 * @version: 1.00.00
 * @desctiption:
 * @company: 厦门立林科技有限公司
 * @author: zhangyx
 * @create: 2024/3/7 11:31
 **/
@EnableKafka
@Configuration
public class OtaKafkaConfig {
    @Resource
    private KafkaProperties kafkaProperties;

    @Bean
    KafkaListenerContainerFactory<ConcurrentMessageListenerContainer<String, String>> otaKafkaBatchListener() {
        ConcurrentKafkaListenerContainerFactory<String, String> factory = this.kafkaListenerContainerFactory();
        factory.setConcurrency(this.kafkaProperties.getListener().getConcurrency());
        return factory;
    }

    private ConcurrentKafkaListenerContainerFactory<String, String> kafkaListenerContainerFactory() {
        ConcurrentKafkaListenerContainerFactory<String, String> factory = new ConcurrentKafkaListenerContainerFactory();
        factory.setConsumerFactory(this.consumerFactory());
        factory.setBatchListener(true);
        factory.getContainerProperties().setPollTimeout(this.kafkaProperties.getListener().getPollTimeout().toMillis());
        factory.getContainerProperties().setAckMode(ContainerProperties.AckMode.MANUAL);
        return factory;
    }

    private ConsumerFactory<String, String> consumerFactory() {
        return new DefaultKafkaConsumerFactory(this.kafkaProperties.buildConsumerProperties());
    }
}
